package com.example.mq2.mqserver.datacenter;

import com.example.mq2.common.MqException;
import com.example.mq2.mqserver.core.Binding;
import com.example.mq2.mqserver.core.Exchange;
import com.example.mq2.mqserver.core.MSGQueue;
import com.example.mq2.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 内存存储中心
 */
public class MemoryDataCenter {

    //key 是 exchangeName
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //key 是 queueName
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //key1 是 exchangeName, key2 是 queueName
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingMap = new ConcurrentHashMap<>();
    //key 是 messageId 消息中心
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //key 是 queueName，List 是对应队列中的消息
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //待确认的消息，key1 是 queueName, key2 是 messageId
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> waitAckMessageMap = new ConcurrentHashMap<>();

    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 交换机添加成功！exchangeName=" + exchange.getName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功！exchangeName=" + exchangeName);
    }

    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 队列添加成功！queueName=" + queue.getName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功！queueName=" + queueName);
    }

    public void insertBinding(Binding binding) throws MqException {
        //查询一下当前 binding 是否存在，不存在就创建
        ConcurrentHashMap<String, Binding> bindingNestMap = bindingMap.computeIfAbsent(binding.getExchangeName(), k ->
                new ConcurrentHashMap<>());
        synchronized (binding) {
            if (bindingNestMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 当前绑定已存在！exchangeName=" + binding.getExchangeName()
                        + ", queueName=" + binding.getQueueName());
            }
            bindingNestMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 添加绑定成功！exchangeName=" + binding.getExchangeName() +
                ", queueName=" + binding.getQueueName());
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingNestMap = bindingMap.get(binding.getExchangeName());
        synchronized (binding) {
            if (bindingNestMap == null) {
                throw new MqException("[MemoryDataCenter] 当前绑定不存在！exchangeName=" + binding.getExchangeName() +
                        ", queueName=" + binding.getQueueName());
            }
            bindingNestMap.remove(binding.getQueueName());
        }
        System.out.println("[MemoryDataCenter] 删除绑定成功！exchangeName=" + binding.getExchangeName() +
                ", queueName=" + binding.getQueueName());
    }

    //获取绑定有两个版本

    /**
     * 获取该交换机的所有绑定
     *
     * @param exchangeName
     * @return
     */
    public ConcurrentHashMap<String, Binding> getBindingMap(String exchangeName) {
        return bindingMap.get(exchangeName);
    }

    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingNestMap = bindingMap.get(exchangeName);
        if (bindingNestMap == null) {
            return null;
        }
        return bindingNestMap.get(queueName);
    }

    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 添加消息成功！messageId=" + message.getMessageId());
    }

    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 删除消息成功！messageId=" + messageId);
    }

    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    /**
     * 发送消息到指定队列
     *
     * @param queue
     * @param message
     */
    public void sendMessage(MSGQueue queue, Message message) {
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k ->
                new LinkedList<>());
        synchronized (queue) {
            messages.add(message);
        }
        addMessage(message);
        System.out.println("[MemoryDataCenter] 成功将消息添加到指定队列中！queueName=" + queue.getName() +
                ", messageId=" + message.getMessageId());
    }

    /**
     * 弹出队首消息
     *
     * @param queue
     * @return
     */
    public Message pollMessage(MSGQueue queue) throws MqException {
        LinkedList<Message> messageList = queueMessageMap.get(queue.getName());
        synchronized (queue) {
            if (messageList == null || messageList.size() == 0) {
                return null;
            }
            Message message = messageList.remove(0);
            System.out.println("[MemoryDataCenter] 成功弹出队列消息! queueName=" + queue.getName());
            return message;
        }
    }

    /**
     * 获取指定队列的消息个数
     * @param queue
     * @return
     */
    public int getMessageCount(MSGQueue queue) {
        LinkedList<Message> messageList = queueMessageMap.get(queue.getName());
        synchronized (queue) {
            if (messageList == null) {
                return 0;
            }
            return messageList.size();
        }
    }

    public void addWaitAckMessage(String queueName, Message message) {
        ConcurrentHashMap<String, Message> waitAckMsgNest = waitAckMessageMap.computeIfAbsent(queueName, k ->
                new ConcurrentHashMap<>());
        waitAckMsgNest.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 成功添加待确认消息！queueName=" + queueName +
                ", messageId=" + message.getMessageId());
    }

    public void removeWaitAckMessage(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> waitAckMsgNest = waitAckMessageMap.get(queueName);
        if (waitAckMsgNest == null) {
            return;
        }
        waitAckMsgNest.remove(messageId);
        System.out.println("[MemoryDataCenter] 成功删除待确认消息！queueName=" + queueName +
                ", messageId=" + messageId);
    }

    public Message getWaitAckMessage(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> waitAckMsgNest = waitAckMessageMap.get(queueName);
        if (waitAckMsgNest == null) {
            return null;
        }
        return waitAckMsgNest.get(messageId);
    }

    /**
     * 将硬盘的数据恢复到内存中
     */
    public void recovery(DiskDataCenter diskDataCenter) throws MqException, IOException {
        //1.交换机
        List<Exchange> exchanges = diskDataCenter.selectAllExchange();
        for(Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }
        //2.队列
        List<MSGQueue> queues = diskDataCenter.selectAllQueue();
        for(MSGQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }
        //3.交换机
        List<Binding> bindings = diskDataCenter.selectAllBinding();
        for(Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingNestMap = bindingMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingNestMap.put(binding.getQueueName(), binding);
        }
        //4.恢复消息数据
        for(MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue);
            queueMessageMap.put(queue.getName(), messages);
            for(Message message: messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }
        System.out.println("[MemoryDataCenter] 内存数据恢复完成！");
        //待确认的消息不用恢复了，硬盘存储的时候也没考虑
    }


}